package org.example.service.impl;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.config.KafkaProcessCacheConfig;
import org.example.exception.NotFoundException;
import org.example.service.KafkaProcessHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @author jingjing.zhang
 */
@Component
@Slf4j
public class KafkaDemoConsumer {
    private static String PUSH_APP_ORDER = "pushAppOrder";
    private static String PUSH_CUSTOM_INFO = "pushCustomInfo";

    @Resource(name = "kafkaProcessCache")
    private KafkaProcessCacheConfig.KafkaProcessCache kafkaProcessCache;
    @Autowired
    private ApplicationContext applicationContext;

    /**
     * 通过监听器的方式,处理kafka消息
     * */
    @KafkaListener(topics = {"hello"},groupId = "test",errorHandler = "consumerAwareErrorHandler")
    public void consumerListener(ConsumerRecord<String,String> record, Acknowledgment ack) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            log.info("Received: " + record);
            try {
                if (kafkaMessage.isPresent()) {
                    String key = record.key();
                    String message = record.value();
                    log.info("Received message={}", message);
                    String beanName = kafkaProcessCache.processCache.get(key);
                    if (StringUtils.isBlank(beanName)){
                        throw new NotFoundException(String.format("key: %s, kafkaProcessHandler is not found",key));
                    }
                    KafkaProcessHandler procecssHandler = (KafkaProcessHandler) applicationContext.getBean(beanName);
                    procecssHandler.handler(message);
                    /**
                     * 手动提交kafka消费完成ACK
                     * */
                    ack.acknowledge();
                }
            } catch (Exception e) {
                log.error("Received message exception",e);
            }
    }
    /**
     * 异常处理
     * */
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, e, consumer) -> {
            log.error("consumerAwareErrorHandler receive : "+message.getPayload().toString());
            log.error("consumerAwareErrorHandler consumer ={}", JSON.toJSONString(consumer));
            log.error("consumerAwareErrorHandler exception : {}",e);
            return null;
        };
    }

}
